package com.atguigu.gmall.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
import com.atguigu.gmall.realtime.app.func.DimSinkFunction;
import com.atguigu.gmall.realtime.app.func.MyDebeziumDeserializationSchema;
import com.atguigu.gmall.realtime.app.func.TableProcessFunction;
import com.atguigu.gmall.realtime.beans.TableProcess;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.flink.util.OutputTag;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;

/**
 * Author: Felix
 * Date: 2022/4/20
 * Desc: 业务数据动态分流
 * 需要启动的进程
 *      zk、kafka、maxwell、hdfs、hbase、BaseDBApp
 * 开发流程
 *      ---------------------业务流数据读取---------------------------
 *      基本环境准备
 *      检查点相关设置
 *      从Kafka的ods_base_db_m主题中读取业务流数据
 *      对读取的数据进行转换  jsonStr->jsonObj
 *      对业务数据进行简单ETL
 *      ---------------------配置流数据读取---------------------------
 *      使用FlinkCDC读取MySQL配置表table_process内容
 *      对读取到的配置流进行广播 ，并创建广播状态
 *      ---------------------将业务流和配置流进行关联---------------------------
 *      业务流.connect(广播配置流)
 *      ---------------------对关联后的流进行处理---------------------------
 *      调用process算子对关联后的数据进行处理----分流
 *      单独定义了一个类TableProcessFunction继承BroadcastProcessFunction
 *          processElement---处理业务流数据
 *              4.在向下游传递数据前，将不需要的字段过滤掉
 *              2.从广播状态中获取当前处理的数据对应的配置信息，进行分流    事实---主流        维度---侧输出流
 *          processBroadcastElement---处理广播流数据
 *              3.提前创建维度表
 *              1.从流中获取配置表中的一条配置信息并封装为TableProcess对象，将配置对象放到广播状态中 Map<sourceTable:operateType,TableProcess>
 *      ---------------------对分流后的数据进行处理---------------------------
 *      将维度侧输出流数据输出到phoenix表中
 *      将主流事实数据输出到kafka不同的主题中
 *
 * 程序执行流程
 *      将需要启动进程都启动起来zk、kafka、maxwell、hdfs、hbase、BaseDBApp
 *      在启动BaseDBApp应用的时候，会从配置表中读取配置信息，并且创建出维度表，将配置信息加载到广播状态中保存起来
 *      对业务数据库的表进行操作
 *          例如：向品牌表base_trademark中添加了一条数据
 *      binlog会记录品牌表的变化
 *      maxwell从binlog中读取变化的数据，并且封装json格式字符串，发送到kafka的ods_base_db_m主题中
 *      BaseDBApp从ods_base_db_m主题中读取数据，进行分流操作
 *          根据当前maxwell封装的json对象，获取操作的业务数据库表名以及操作类型
 *          将表名和操作类型拼接为key，用这个key到广播状态中获取对应的配置信息
 *          如果有配置信息，根据配置信息中的sinkType进行分流
 *              在分流前，做了两件比较重要的事
 *                  补充sink_table，输出的目的地
 *                  字段的过滤
 *              主流----事实
 *              侧输出流-维度
 *          如果没有配置信息，说明在配置表中，没有对当前操作进行配置，直接过滤掉。
 *      将维度侧输出流数据输出到phoenix表中
 *      将主流事实数据输出到kafka不同的主题中
 */
public class BaseDBApp {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        /*
        //TODO 2.检查点相关的设置
        //2.1 开启检查点
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        //2.2 设置检查点超时时间
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        //2.3 设置job取消后，检查点是否保留
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //2.4 设置两个检查点之间最小时间间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L);
        //2.5 设置重启策略
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30),Time.seconds(3)));
        //2.6 设置状态后端
        env.setStateBackend(new FsStateBackend("hfds://hadoop202:8020/xxxx"));
        //2.7 设置操作hadoop的用户
        System.setProperty("HADOOP_USER_NAME","atguigu");
        */
        //TODO 3.从kafka中读取业务数据
        //3.1 声明消费的主题以及消费者组
        String topic = "ods_base_db_m";
        String groupId = "base_db_group";
        //3.2 获取消费者对象
        FlinkKafkaConsumer<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupId);
        //3.3 消费数据  封装为流
        DataStreamSource<String> kafkaStrDS = env.addSource(kafkaSource);

        //TODO 4.对读取到的数据进行类型转换     jsonStr->jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.map(JSON::parseObject);

        //TODO 5.对读取到的业务数据进行简单的ETL操作
        SingleOutputStreamOperator<JSONObject> filterDS = jsonObjDS.filter(
            new FilterFunction<JSONObject>() {
                @Override
                public boolean filter(JSONObject jsonObj) throws Exception {
                    boolean flag = jsonObj.getString("table") != null
                        && jsonObj.getString("table").length() != 0
                        && jsonObj.getJSONObject("data") != null
                        && jsonObj.getString("data").length() > 3;
                    return flag;
                }
            }
        );

        //filterDS.print(">>>>>>");

        //TODO 6.使用FlinkCDC读取配置表数据--配置流
        DebeziumSourceFunction<String> mySqlSource = MySQLSource.<String>builder()
            .hostname("hadoop202")
            .port(3306)
            .databaseList("gmall1018_realtime")
            .tableList("gmall1018_realtime.table_process")
            .username("root")
            .password("123456")
            .deserializer(new MyDebeziumDeserializationSchema())
            .startupOptions(StartupOptions.initial())
            .build();

        DataStreamSource<String> mySqlDS = env.addSource(mySqlSource);

        //mySqlDS.print(">>>>");

        //TODO 7.将配置流数据进行广播--广播流
        MapStateDescriptor<String, TableProcess> mapStateDescriptor
            = new MapStateDescriptor<String, TableProcess>("mapStateDescriptor", String.class, TableProcess.class);
        BroadcastStream<String> broadcastDS = mySqlDS.broadcast(mapStateDescriptor);

        //TODO 8.将业务流和广播流进行关联       connect
        BroadcastConnectedStream<JSONObject, String> connectDS = filterDS.connect(broadcastDS);

        //TODO 9.对关联之后的数据进行处理  分流   事实数据---主流       维度数据---侧输出流
        OutputTag<JSONObject> dimTag = new OutputTag<JSONObject>("dimTag") {
        };

        SingleOutputStreamOperator<JSONObject> realDS = connectDS.process(
            new TableProcessFunction(dimTag, mapStateDescriptor)
        );

        DataStream<JSONObject> dimDS = realDS.getSideOutput(dimTag);
        realDS.print(">>>>");
        dimDS.print("####");

        //TODO 10.将维度侧输出流数据输出到phoenix的维度表中
        dimDS.addSink(new DimSinkFunction());

        //TODO 11.将主流中的事实数据输出到kafka不同的主题中
        realDS.addSink(
            MyKafkaUtil.getKafkaSinkBySchema(new KafkaSerializationSchema<JSONObject>() {
                @Override
                public ProducerRecord<byte[], byte[]> serialize(JSONObject jsonObj, @Nullable Long timestamp) {
                    String topic = jsonObj.getString("sink_table");
                    return new ProducerRecord<byte[], byte[]>(topic,jsonObj.getJSONObject("data").toJSONString().getBytes());
                }
            })
        );

        env.execute();

    }
}
